/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.delivery;

import lk.mrt.cse.wihidum.communicator.*;
import lk.mrt.cse.wihidum.processor.PublicationMessage;
import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.util.*;
import org.apache.axiom.om.*;
import org.apache.axiom.soap.*;
import org.apache.axis2.AxisFault;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.*;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.transport.http.HTTPConstants;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.httpclient.HttpClient;
import org.apache.log4j.Logger;

import javax.xml.namespace.QName;
import java.util.*;

/**
 * Thread to handle a single publication
 */
public class PublicationThread
        implements Runnable
{

    private static final Logger logger = Logger.getLogger(PublicationThread.class);
    private PublicationMessage publicationMessage;
    private Router router;
    private HashMap<String, String> brokerMap;
    private HttpClient httpClient;

    /**
     * Default constructor
     *
     * @param publicationMessage PublicationMessage
     * @param brokerMap BrokerMap
     */
    public PublicationThread(PublicationMessage publicationMessage, HashMap<String, String> brokerMap)
    {
        this.publicationMessage = publicationMessage;
        this.brokerMap = brokerMap;
        httpClient = new HttpClient();
    }

    /**
     * Send the publication to interested subscribers
     */
    public void run()
    {


        if (brokerMap.get("topology").equals(WihidumConstants.TopologyNames.clusterTopology))
        {
            router = new ClusterRouter();
            router.propagatePublication(publicationMessage);
            this.deliverPublication();
            this.deliverBrokerPublication();
        }
        if (brokerMap.get("topology").equals(WihidumConstants.TopologyNames.treeTopology))
        {
            router = new TreeRouter();
            this.deliverPublication();
            this.deliverBrokerPublication();
        }

    }

    private void deliverPublication()
    {
        try
        {
            ServiceClient serviceClient;
            serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(), null);

            //Delivering publication to normal subscribers
            ArrayList<Subscriber> subscriberList = SubscriptionMap.getSubscribersFilteredByTopic(publicationMessage.getTopic());

            if(subscriberList != null && !subscriberList.isEmpty())
            {
            for (Subscriber subscriber : subscriberList)
            {
                //creating in out operating client
                OperationClient operationClient = serviceClient.createClient(ServiceClient.ANON_OUT_ONLY_OP);

                //creating message context
                MessageContext outMessageContext = new MessageContext();
                Options options = outMessageContext.getOptions();
                options.setProperty(HTTPConstants.CACHED_HTTP_CLIENT, httpClient);
                options.setProperty(HTTPConstants.REUSE_HTTP_CLIENT, Boolean.TRUE);

                String SOAPVersion = options.getSoapVersionURI();
                if (SOAPVersion == null)
                    SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

                EndpointReference targetEPR = new EndpointReference(subscriber.getNotifyToAddress());
                options.setTo(targetEPR);
                options.setAction(WihidumConstants.Operations.STORE_PUBLICATION);

                SOAPEnvelope deliveryEnvelope = this.createPublicationDeliveryMessage(SOAPVersion);

                outMessageContext.setEnvelope(deliveryEnvelope);
                operationClient.addMessageContext(outMessageContext);
                operationClient.execute(true);

                logger.debug("Publication delivery process completed successfully");
            }
             logger.info("Event#"+ publicationMessage.getMessageID() +" was routed successfully");
            }
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing store publication response", ex);
        }
        catch (AxisFault ex)
        {
            logger.error("Error in sending publication", ex);
        }
    }

    private void deliverBrokerPublication()
    {
        try
        {
            ServiceClient serviceClient;
            serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(), null);

            //Delivering publication to broker subscribers
            ArrayList<BrokerSubscription> brokerSubscriptionList = BrokerSubscriptionMap.getBrokerSubscribersFilteredByTopic(publicationMessage.getTopic());
           if(brokerSubscriptionList != null && !brokerSubscriptionList.isEmpty())
           {
            for (BrokerSubscription brokerSubscription : brokerSubscriptionList)
            {
                logger.debug("Preparing to send publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                if (!brokerSubscription.getFromBrokerAddress().equals(publicationMessage.getBrokerPublicationElementFrom()))
                {
                    //creating in out operating client
                    OperationClient operationClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);

                    //creating message context
                    MessageContext outMessageContext = new MessageContext();
                    Options options = outMessageContext.getOptions();

                    options.setProperty(HTTPConstants.CACHED_HTTP_CLIENT, httpClient);
                    options.setProperty(HTTPConstants.REUSE_HTTP_CLIENT, Boolean.TRUE);

                    String SOAPVersion = options.getSoapVersionURI();
                    if (SOAPVersion == null)
                        SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

                    options.setTo(new EndpointReference(brokerSubscription.getFromBrokerAddress()));
                    options.setAction(WihidumConstants.Operations.BROKER_PUBLISH);

                    NeighbourBroker neighbourBroker = new NeighbourBroker();
                    neighbourBroker.setAddress(brokerSubscription.getFromBrokerAddress());

                    neighbourBroker.setBroekrID(brokerSubscription.getFromBrokerID());

                    SOAPEnvelope brokerPublicationMessage = router.createPublicationRequest(publicationMessage, neighbourBroker, SOAPVersion);

                    outMessageContext.setEnvelope(brokerPublicationMessage);
                    operationClient.addMessageContext(outMessageContext);
                    logger.debug("Sending publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                    operationClient.execute(true);

                    MessageContext inMessageContext = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

                    SOAPEnvelope response = inMessageContext.getEnvelope();

                    logger.debug("Publication delivery response: " + this.processBrokerPublicationResponse(response));
                }
                else
                {
                    logger.debug("This publication has arrived from the same broker subscriber : " + brokerSubscription.getFromBrokerAddress() + " to which we are trying to diliver the broker publication. Hence, the delivery was cancelled.");
                }
            }
            logger.info("Event#"+ publicationMessage.getMessageID() +" was routed successfully");
        }
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing store publication response", ex);
        }
        catch (AxisFault ex)
        {
            logger.error("Error in sending publication", ex);
        }
    }

    /*
   /**
    * Create element to be sent to the listener
    *
    * @param SOAPURI SOAP version to be used in delivery message
    * @return Delivery element
    */
    private SOAPEnvelope createPublicationDeliveryMessage(String SOAPURI)
    {
        try
        {
            SOAPFactory soapFactory = null;
            if (SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP11Factory();
            else if (SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP12Factory();
            else throw new Exception("Unknown SOAP version");

            SOAPEnvelope soapEnvelope = soapFactory.getDefaultEnvelope();

            //declare the used namespaces in the SOAP Envelope
            OMNamespace wseventingNS = soapFactory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
            soapEnvelope.declareNamespace(wseventingNS);
            OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
            soapEnvelope.declareNamespace(wsaNS);
            OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
            soapEnvelope.declareNamespace(wihidumNS);

            SOAPHeader soapHeader = soapEnvelope.getHeader();

            OMFactory factory = OMAbstractFactory.getOMFactory();


            OMElement publicationDeliveryHeader = factory.createOMElement("PublicationDeliveryHeader", wihidumNS);

            OMElement subscriptionIDElement = factory.createOMElement("SubscriptionID", wihidumNS);
            OMElement publisherAddressElement = factory.createOMElement("PublisherAddress", wihidumNS);
            OMElement topicElement = factory.createOMElement("Topic", wihidumNS);

            publisherAddressElement.setText(publicationMessage.getEventSourceURL());
            topicElement.setText(publicationMessage.getTopic());

            publicationDeliveryHeader.addChild(subscriptionIDElement);
            publicationDeliveryHeader.addChild(publisherAddressElement);
            publicationDeliveryHeader.addChild(topicElement);

            soapHeader.addChild(publicationDeliveryHeader);
            SOAPBody soapBody = soapEnvelope.getBody();
            soapBody.setText(publicationMessage.getBody());

            return soapEnvelope;
        }
        catch (Exception ex)
        {
            logger.error(ex);
            return null;
        }

    }

    //TODO: Attach error channel
    private String processBrokerPublicationResponse(SOAPEnvelope responseEnvelope)
    {
        String responseString = "";
         try
         {
            if (responseEnvelope != null)
            {
                OMElement responseElement = responseEnvelope.getBody().getFirstElement();
                OMElement referenceParameters = responseElement.getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "ReferenceParameters"));
                OMElement sourceEPRElement = referenceParameters.getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "SourceEPR"));
                responseString = sourceEPRElement.getText();
            }
            else
                responseString = "Error";
         }
         catch (NullPointerException ex)
         {
             ex.printStackTrace();
             responseString = "Error";
         }
        return responseString;
    }
}
